{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "885b3341",
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "import asyncio\n",
    "from mcp import ClientSession, StdioServerParameters\n",
    "from mcp.client.stdio import stdio_client\n",
    "import os\n",
    "from dotenv import load_dotenv\n",
    "from agents import Agent, Runner, trace, Tool\n",
    "from agents.mcp import MCPServerStdio\n",
    "from IPython.display import Markdown, display\n",
    "from datetime import datetime\n",
    "import re\n",
    "import shutil\n",
    "import time\n",
    "\n",
    "load_dotenv(override=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "01d11844",
   "metadata": {},
   "outputs": [],
   "source": [
    "def extract_latest_screenshot_path(result):\n",
    "    \"\"\"\n",
    "    Extract the latest (most recent) screenshot file path from an MCP result object.\n",
    "\n",
    "    Supports:\n",
    "    - structuredContent[\"path\"]\n",
    "    - meta[\"path\"]\n",
    "    - Any path inside text content (e.g. \"/tmp/playwright-mcp-output/...png\")\n",
    "\n",
    "    Returns:\n",
    "        str: The most recent screenshot path found.\n",
    "    Raises:\n",
    "        ValueError if no valid path is found.\n",
    "    \"\"\"\n",
    "    paths = []\n",
    "\n",
    "    # 1️⃣ structuredContent\n",
    "    if hasattr(result, \"structuredContent\") and isinstance(result.structuredContent, dict):\n",
    "        path = result.structuredContent.get(\"path\")\n",
    "        if path:\n",
    "            paths.append(path)\n",
    "\n",
    "    # 2️⃣ meta\n",
    "    if hasattr(result, \"meta\") and isinstance(result.meta, dict):\n",
    "        path = result.meta.get(\"path\")\n",
    "        if path:\n",
    "            paths.append(path)\n",
    "\n",
    "    # 3️⃣ text-based path detection in content\n",
    "    if hasattr(result, \"content\") and result.content:\n",
    "        for item in result.content:\n",
    "            if hasattr(item, \"text\") and item.text:\n",
    "                # Find all paths that look like /tmp/playwright-mcp-output/...png\n",
    "                found = re.findall(r\"/tmp/playwright-[\\w/-]+\\.png\", item.text)\n",
    "                paths.extend(found)\n",
    "\n",
    "    # 4️⃣ Clean up duplicates\n",
    "    paths = list(dict.fromkeys(paths))\n",
    "\n",
    "    if not paths:\n",
    "        raise ValueError(f\"Could not determine screenshot path from result: {result}\")\n",
    "\n",
    "    # 5️⃣ Return the latest (last one in the list)\n",
    "    latest_path = paths[-1]\n",
    "    print(f\"path : {latest_path}\")\n",
    "    return latest_path"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "64dac4ba",
   "metadata": {},
   "outputs": [],
   "source": [
    "async def take_and_save_screenshot(session, label: str):\n",
    "    \"\"\"\n",
    "    Take a screenshot via Playwright MCP and save it to a custom folder with a given label.\n",
    "    Returns the final saved path.\n",
    "    \"\"\"\n",
    "    await asyncio.sleep(0.5)\n",
    "\n",
    "    # Call the MCP tool\n",
    "    print(\"call mcp tool\")\n",
    "    result = await session.call_tool(\"browser_take_screenshot\", {})\n",
    "\n",
    "    # Try to find the actual temp screenshot path\n",
    "    print(\"Construct path\")\n",
    "    tmp_path = extract_latest_screenshot_path(result)\n",
    "    if not os.path.exists(tmp_path):\n",
    "        raise FileNotFoundError(f\"Screenshot file not found: {tmp_path}\")\n",
    "\n",
    "    # Prepare custom folder\n",
    "    print(\"Prepare customer folder\")\n",
    "    base_dir = os.path.join(os.getcwd(), \"screenshots\")\n",
    "    os.makedirs(base_dir, exist_ok=True)\n",
    "\n",
    "    # Create a unique name\n",
    "    timestamp = time.strftime(\"%Y%m%d-%H%M%S\")\n",
    "    filename = f\"{label}_{timestamp}.png\"\n",
    "    dest_path = os.path.join(base_dir, filename)\n",
    "\n",
    "    # Copy file from MCP’s temp folder\n",
    "    print(\"Ready to copy\")\n",
    "    shutil.copy(tmp_path, dest_path)\n",
    "    print(f\"📸 Saved screenshot: {dest_path}\")\n",
    "\n",
    "    return dest_path"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2242e9a0",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Test Playwright MCP Server\n",
    "async def test_playwright_server():\n",
    "    \n",
    "    # Connect to Playwright MCP server\n",
    "    server_params = StdioServerParameters(\n",
    "        command=\"npx\",\n",
    "        args=[\"@playwright/mcp@latest\", \"--browser\", \"chromium\"],\n",
    "        env={\"DISPLAY\": os.environ.get(\"DISPLAY\", \":0\")}\n",
    "    )\n",
    "    \n",
    "    async with stdio_client(server_params) as (read, write):\n",
    "        async with ClientSession(read, write) as session:\n",
    "            await session.initialize()\n",
    "            \n",
    "            # List available tools\n",
    "            print(\"📋 Available Playwright Tools:\")\n",
    "            tools_response = await session.list_tools()\n",
    "            for tool in tools_response.tools:\n",
    "                print(f\"  - {tool.name}\")\n",
    "            \n",
    "            print(\"\\n🎭 Testing Playwright Automation:\")\n",
    "            \n",
    "            # Test 1: Navigate to a webpage\n",
    "            print(\"\\n1. Navigating to example.com...\")\n",
    "            result1 = await session.call_tool(\"browser_navigate\", {\n",
    "                \"url\": \"https://example.com\"\n",
    "            })\n",
    "            print(f\"   Result: {result1.content[0].text[:200] if result1.content else 'Success'}\")\n",
    "            \n",
    "            # Test 2: Take a screenshot\n",
    "            print(\"Before taking screenshot1\")\n",
    "            \n",
    "            dest_path = await take_and_save_screenshot(session, \"test_screenshot.png\")\n",
    "            print(f\" Screenshot saved at: {dest_path} \")\n",
    "            #screenshot_path = os.path.join(screenshot_dir, \"test_screenshot.png\")\n",
    "            #print(f\"\\n2. Taking screenshot to: {screenshot_path}\")\n",
    "            #result2 = await session.call_tool(\"browser_take_screenshot\", {\n",
    "            #    \"path\": screenshot_path\n",
    "            #})\n",
    "            #print(f\"   Result: {result2.content[0].text if result2.content else 'Success'}\")\n",
    "            \n",
    "            # Verify file exists\n",
    "            #if os.path.exists(screenshot_path):\n",
    "            #    print(f\"   ✅ Screenshot saved! Size: {os.path.getsize(screenshot_path)} bytes\")\n",
    "            #else:\n",
    "            #    print(f\"   ❌ Screenshot not found at: {screenshot_path}\")\n",
    "            \n",
    "            # Test 3: Navigate to todo app\n",
    "            print(\"\\n3. Navigating to todo app...\")\n",
    "            result3 = await session.call_tool(\"browser_navigate\", {\n",
    "                \"url\": \"https://eviltester.github.io/simpletodolist/todo.html\"\n",
    "            })\n",
    "            print(f\"   Loaded todo app {result3}\")\n",
    "\n",
    "            # Wait for the input to be visible (optional, but good practice)\n",
    "            await session.call_tool(\"browser_wait_for\", {\n",
    "                \"selector\": \".new-todo\",\n",
    "                \"state\": \"visible\",\n",
    "                \"timeout\": 5000\n",
    "            })\n",
    "\n",
    "            print(\"\\n4. Typing into todo input...\")\n",
    "            result4 = await session.call_tool(\"browser_type\", {\n",
    "                \"element\": \"Enter new todo text here\",\n",
    "                \"ref\": \"e8\",\n",
    "                \"text\": \"Test task from Jupyter\"\n",
    "            })\n",
    "            print(f\"   Input typed! {result4}\")\n",
    "\n",
    "            \"\"\"\n",
    "            # Step 2. Take a fresh DOM snapshot (important!)\n",
    "            await session.call_tool(\"browser_snapshot\", {})\n",
    "            \n",
    "            # Test 4: Fill a form field\n",
    "            print(\"\\n4. Filling todo input...\")\n",
    "            result4 = await session.call_tool(\"browser_fill_form\", {\n",
    "                \"fields\": [\n",
    "                    {\n",
    "                        \"name\": \"newTodo\",\n",
    "                        \"type\": \"textbox\",\n",
    "                        \"ref\": \".new-todo\",  # or input.new-todo\n",
    "                        \"value\": \"Test task from Jupyter\"\n",
    "                    }\n",
    "                ]\n",
    "            })\n",
    "            print(f\"   Input filled! {result4}\")\n",
    "            \"\"\"\n",
    "            await take_and_save_screenshot(session, \"test_todo_input_filled.png\")\n",
    "            \n",
    "            # Test 5: Press Enter key\n",
    "            print(\"\\n5. Pressing Enter key...\")\n",
    "            result5 = await session.call_tool(\"browser_press_key\", {\n",
    "                \"key\": \"Enter\"\n",
    "            })\n",
    "            print(f\"   Key pressed!\")\n",
    "            \n",
    "            # Test 6: Take final screenshot\n",
    "            dest_path = await take_and_save_screenshot(session, \"test_todo_screenshot.png\")\n",
    "            print(f\" Screenshot saved at: {dest_path} \")\n",
    "            \"\"\"\n",
    "            final_screenshot = os.path.join(screenshot_dir, \"test_todo_screenshot.png\")\n",
    "            print(f\"\\n6. Taking final screenshot to: {final_screenshot}\")\n",
    "            result6 = await session.call_tool(\"browser_take_screenshot\", {\n",
    "                \"path\": final_screenshot\n",
    "            })\n",
    "            \n",
    "            if os.path.exists(final_screenshot):\n",
    "                print(f\"   ✅ Final screenshot saved! Size: {os.path.getsize(final_screenshot)} bytes\")\n",
    "            else:\n",
    "                print(f\"   ❌ Screenshot not found!\")\n",
    "            \"\"\"\n",
    "            \n",
    "            # Test 7: Get browser snapshot (page info)\n",
    "            print(\"\\n7. Getting page snapshot...\")\n",
    "            result7 = await session.call_tool(\"browser_snapshot\", {})\n",
    "            print(f\"   Snapshot: {result7.content[0].text[:300] if result7.content else 'No content'}...\")\n",
    "            \n",
    "            print(\"\\n✅ All Playwright tests completed!\")\n",
    "            #print(f\"📂 Check screenshots in: {screenshot_dir}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2c2b86e6",
   "metadata": {},
   "outputs": [],
   "source": [
    "await test_playwright_server()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5d4dc2af",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
